package com.spark.lola.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.spark.lola.logistics.entity.CarSubscribe;
import com.spark.lola.logistics.entity.Message;
import com.spark.lola.logistics.entity.OrderInfo;
import com.spark.lola.logistics.entity.OrderProcess;
import com.spark.lola.logistics.service.MessageService;
import com.spark.lola.support.common.constant.CommonConstant;
import com.spark.lola.support.common.entity.ShiroUser;
import com.spark.lola.support.netty.common.NettyCacheUtils;
import com.spark.lola.support.netty.service.HawkPushServiceApi;
import com.spark.lola.support.wms.cache.CollectionCache;
import com.spark.lola.support.wms.configuration.JPushProperties;
import com.spark.lola.support.wms.configuration.WmsProperties;
import com.spark.lola.support.wms.constant.KafkaTopic;
import com.spark.lola.support.wms.constant.WmsNettyCommands;
import com.spark.lola.support.wms.dto.JPushSendDto;
import com.spark.lola.support.wms.dto.OrderDto;
import com.spark.lola.support.wms.util.JPushUtil;
import io.netty.channel.Channel;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * @author: Zhi-peng wang
 * @date: 2020/05/27 14:39
 **/
@Component
public class CarSubscribeConsumer {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private WmsProperties wmsProperties;

    @Autowired
    private HawkPushServiceApi pushServiceApi;

    @Autowired
    private MessageService messageService;

    @Autowired
    private JPushProperties jPushProperties;

    @Autowired
    private JPushUtil jPushUtil;

    @KafkaListener(topics = KafkaTopic.LOGISTICS_CAR_SUBSCRIBE_CHANGE)
    public void change(ConsumerRecord<String, String> consumerRecord) {
        OrderDto<JSONObject> model = JSON.parseObject(consumerRecord.value(), OrderDto.class);
        CarSubscribe carSubscribe = JSON.parseObject(model.getOrder().toJSONString(), CarSubscribe.class);
        int type = Integer.parseInt(consumerRecord.key());
        // 极光推送给预订人
        JPushSendDto sendDtoSubscribe = new JPushSendDto();
        sendDtoSubscribe.setTitle("回头车预定");
        sendDtoSubscribe.setContent(model.getJPushMessage());
        sendDtoSubscribe.setPushConditions(Collections.singletonList(carSubscribe.getSubscribeId().toString()));
        jPushUtil.sendPushWithCallback(jPushProperties.getPurchase(), sendDtoSubscribe);
        // 写入消息
        messageService.save(new Message(carSubscribe.getSubscribeId(), "车牌：" + carSubscribe.getCarNo(), model.getJPushMessage()));
        // 审核通过和取消预订推送给驾驶员
        if (type == 1) {
            String pushMessage = String.format("%s预订了你的回头车", carSubscribe.getSubscribeName());
            JPushSendDto sendDtoDriver = new JPushSendDto();
            sendDtoDriver.setTitle("回头车预定");
            sendDtoDriver.setContent(pushMessage);
            sendDtoDriver.setPushConditions(Collections.singletonList(carSubscribe.getDrivingId().toString()));
            jPushUtil.sendPushWithCallback(jPushProperties.getLogistics(), sendDtoDriver);
            // 写入消息
            messageService.save(new Message(carSubscribe.getDrivingId(), "车牌：" + carSubscribe.getCarNo(), pushMessage));
        }
        if (type == 3) {
            String pushMessage = String.format("%s取消预订了你的回头车", carSubscribe.getSubscribeName());
            JPushSendDto sendDtoDriver = new JPushSendDto();
            sendDtoDriver.setTitle("回头车预定");
            sendDtoDriver.setContent(pushMessage);
            sendDtoDriver.setPushConditions(Collections.singletonList(carSubscribe.getDrivingId().toString()));
            jPushUtil.sendPushWithCallback(jPushProperties.getLogistics(), sendDtoDriver);
            // 写入消息
            messageService.save(new Message(carSubscribe.getDrivingId(), "车牌：" + carSubscribe.getCarNo(), pushMessage));
        }

        logger.info("===推送消息");
        dealPush(carSubscribe, model.getMessage());
    }

    private void dealPush(CarSubscribe carSubscribe, String message) {
        Set<Channel> channelSet = new HashSet<>();
        // 推送给超级管理员
        List<ShiroUser> supers = CollectionCache.dutyEmployees.get(1L);
        if (!CollectionUtils.isEmpty(supers)) {
            supers.forEach(superUser -> {
                Set<Channel> channels = NettyCacheUtils.getChannel(superUser.getId().toString());
                if (!CollectionUtils.isEmpty(channels)) {
                    channelSet.addAll(channels);
                }
            });
        }
        // 推送给物流主管
        List<ShiroUser> logistics = CollectionCache.dutyEmployees.get(wmsProperties.getLogistics().getDutyId().get(0));
        if (!CollectionUtils.isEmpty(logistics)) {
            logistics.forEach(lead -> {
                Set<Channel> leadChannelSet = NettyCacheUtils.getChannel(lead.getId().toString());
                if (!CollectionUtils.isEmpty(leadChannelSet)) {
                    channelSet.addAll(leadChannelSet);
                }
            });
        }

        // 推送给预订人
        Set<Channel> subscribeChannelSet = NettyCacheUtils.getChannel(String.valueOf(carSubscribe.getSubscribeId()));
        if (!CollectionUtils.isEmpty(subscribeChannelSet)) {
            channelSet.addAll(subscribeChannelSet);
        }
        if (!CollectionUtils.isEmpty(channelSet)) {
            pushServiceApi.pushMsg(channelSet, WmsNettyCommands.PUSH_CAR_SUBSCRIBE, message);
        }
    }

}
